package com.qupeng.demo.kafka.kafkaapache.producer;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class AvroProducer {
    public void produce() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props. put("key.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
        props.put("value.serializer", "io.confluent.kafka.serializers.KafkaAvroSerializer");
        props.put("schema.registry.url", "localhost:8081");
        Producer<String, Product> producer = new KafkaProducer(props);
        while (true) {
            Product product = Product.newBuilder().build();
            ProducerRecord<String, Product> record = new ProducerRecord("product", product.getName(), product);
            producer.send(record);
        }
    }

    public void produceWithGenericRecord() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props. put("key.serializer", "io.confluent.kafka.serializer.KafkaAvroSerializer");
        props.put("value.serializer", "io.confluent.kafka.serializer.KafkaAvroSerializer");
        props.put("schema.registry.url", "...");
        Producer<String, GenericRecord> producer = new KafkaProducer(props);

        String schemaString = "{\n" +
                "  \"namespace\": \"com.qupeng.demo.kafka.kafkaspringbootproducer.avro\",\n" +
                "  \"type\": \"record\",\n" +
                "  \"name\": \"Product\",\n" +
                "  \"fields\": [\n" +
                "    {\n" +
                "      \"type\": \"int\",\n" +
                "      \"name\": \"id\",\n" +
                "      \"default\": 0\n" +
                "    },\n" +
                "    {\n" +
                "      \"type\": \"string\",\n" +
                "      \"name\": \"name\",\n" +
                "      \"default\": \"\"\n" +
                "    }\n" +
                "  ]\n" +
                "}";
        Schema.Parser parser = new Schema.Parser();
        Schema schema = parser.parse(schemaString);

        GenericRecord genericRecord = new GenericData.Record(schema);
        genericRecord.put("id", 0);
        genericRecord.put("name", "iPhone 17");

        ProducerRecord<String, GenericRecord> producerRecord = new ProducerRecord<>("product", "0", genericRecord);
        producer.send(producerRecord);
    }
}
